package com.ww.flink

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.common.serialization.StringSerializer

import java.text.SimpleDateFormat
import java.util
import java.util.Properties
import scala.collection.JavaConverters._

/**
 * 统计出每一辆车的运行轨迹
 *
 * 1、拿到每辆车的所有信息（车牌号、卡口号、eventtime、speed）
 * 2、根据每辆车 分组
 * 3、对每组数据钟的信息按照eventtime排序 升序   卡口连接起来  轨迹
 *
 */
object Flink_try10_list_state {
  case class CarInfo(carId:String,speed:Long)
  def main(args: Array[String]): Unit = {
    val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:sss")
    // 读取kafka消息
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "192.168.102.19:9092")
    properties.setProperty("key.serializer", classOf[StringSerializer].getName)
    properties.setProperty("value.serializer", classOf[StringSerializer].getName)

    //环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //从kafka读取数据
    val kafkaStream: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String]("flink-kafka", new SimpleStringSchema(), properties))
    //得到 (卡口号,车牌,时间)
    //'310999003702', '沪M05016','2014-08-20 14:09:35',10
    kafkaStream.map(_.split("\t")).map(arr=>{
      val time = format.parse(arr(2)).getTime
      (arr(0),arr(1),time)
    })
    //根据车牌分组 '310999003702', '沪M05016','2014-08-20 14:09:35'
      .keyBy(_._2)
    //用ListState记录卡扣号,并且用时间升序
      .map(new RichMapFunction[(String,String,Long),(String,String)] {
        private var mids:ListState[(String,Long)] = _


        override def open(parameters: Configuration) = {
          val des = new ListStateDescriptor[(String, Long)]("list", createTypeInformation[(String, Long)])
          mids = getRuntimeContext.getListState(des)
        }

        override def map(in: (String, String, Long)) = {
          mids.add((in._1,in._3))
          val midis: Iterator[(String, Long)] = mids.get().asScala.toList.sortBy(_._2).toIterator
          val buffer = new StringBuffer()
          while(midis.hasNext){
            buffer.append(midis.next()._1).append(" ")
          }
          (in._2,buffer.toString)
        }
      }).print()
    //打印出各个车牌的轨迹
    env.execute()
  }
}
